package com.RPC;

import com.one2many.ConnectionUtil;
import com.rabbitmq.client.*;
import com.rabbitmq.client.impl.AMQBasicProperties;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class RPCClient {
    private Connection connection;
    private Channel channel;
    private String rpc_queue_name = "rpc_queue";
    private String replyQueueName;

    public RPCClient() throws Exception {
        connection = ConnectionUtil.getConnection();
        channel = connection.createChannel(1);
        replyQueueName = channel.queueDeclare().getQueue();
    }

    public String call(String message) throws Exception {
        final String corrId = UUID.randomUUID().toString();
        AMQP.BasicProperties props = new AMQP.BasicProperties().builder().correlationId(corrId).replyTo(replyQueueName).build();
        channel.basicPublish("", rpc_queue_name, props, message.getBytes("UTF-8"));
        final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
        //一个客户端向服务器发送请求，服务器端处理请求后，
        // 将其处理结果保存在一个存储体中。
        // 而客户端为了获得处理结果，
        // 那么客户在向服务器发送请求时，同时发送一个回调队列地址reply_to
        channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //Correlation id 关联标识
                //一个客户端可能会发送多个请求给服务器，
                // 当服务器处理完后，客户端无法辨别在回调队列中的响应具体和那个请求时对应的。
                // 为了处理这种情况，客户端在发送每个请求时，
                // 同时会附带一个独有correlation_id属性，
                // 这样客户端在回调队列中根据correlation_id字段的值就可以分辨此响应属于哪个请求。
                if (properties.getCorrelationId().equals(corrId)) {
                    response.offer(new String(body, "UTF-8"));
                }
            }
        });
        return response.take();
    }

    public void close() throws IOException {
        connection.close();
    }
}
